package com.example.loader;

import com.hazelcast.collection.IQueue;
import com.hazelcast.collection.ItemEvent;
import com.hazelcast.collection.ItemListener;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.HazelcastInstanceNotActiveException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

/*
 * @className: HazelcastQueueLoader
 * @author: chentao
 * @date: 2025/2/19 上午8:44
 * @Version: 1.0
 * @description:
 */
@Service
public class HazelcastQueueLoader implements HazelcastLoader {

    Logger logger = LoggerFactory.getLogger(getClass());

    public static final String QUEUE_NAME_KEY = "queue";

    private IQueue<String> queue = null;

    private Thread thread = null;

    private volatile boolean running = true;

    /***
     * 加载queue信息
     * @param hazelcastInstance
     * @throws Exception
     */
    @Override
    public void init(HazelcastInstance hazelcastInstance) throws Exception {
        this.queue = hazelcastInstance.getQueue(QUEUE_NAME_KEY);
        this.queue.addItemListener(new ItemListener() {
            @Override
            public void itemAdded(ItemEvent item) {
                logger.warn("TestQueueController itemAdded ..." + item.getItem());
            }

            @Override
            public void itemRemoved(ItemEvent item) {
                logger.warn("TestQueueController itemRemoved ..." + item.getItem());
            }
        }, true);
        this.thread = new Thread(new Runnable() {
            @Override
            public void run() {
                logger.warn("start queue listener ...");
                while (running) {
                    try {
                        String msg = queue.take();
                        logger.warn("queue收到消息: " + msg);
                    } catch (Exception e) {
                        if (!(e instanceof HazelcastInstanceNotActiveException)) {
                            logger.warn("queue监听出错: " + e.getMessage());
                        }
                    }
                }
            }
        });
        this.thread.start();
    }

    @Override
    public void destroy(HazelcastInstance hazelcastInstance) throws Exception {
        this.running = false;
        if (this.thread.isAlive()) {
            this.thread.interrupt();
        }
    }

    /***
     * 对外接口提供IQueue信息
     * @return
     */
    public IQueue<String> getQueue() {
        return queue;
    }
}
